package com.atguigu.gmall.realtime.app;

import com.atguigu.gmall.realtime.util.FlinkSourceUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import static org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION;

/**
 * @Author lzc
 * @Date 2023/2/8 13:53
 */
public abstract class BaseApp {
    
    protected abstract void handle(StreamExecutionEnvironment env,
                                   DataStreamSource<String> stream);
    
    public void init(int port, int p, String ckAndGroupIdAndJobName, String topic) {
        
        System.setProperty("HADOOP_USER_NAME", "atguigu");
        // 1. 读取数据 ods_db 数据
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", port);
        // 给 job 设置 jobName
        conf.setString("pipeline.name", ckAndGroupIdAndJobName);
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.getCheckpointConfig().enableUnalignedCheckpoints();
        // 设置并行度: 保持和 kafka 的分区数一致
        env.setParallelism(p);
        // 开启 checkpoint
        // 1. 设置状态后端: HashMapStateBackend 和 rocksdb
        env.setStateBackend(new HashMapStateBackend());
        // 2. 开启 checkpoint
        env.enableCheckpointing(3000);
        // 3. 设置 checkpoint 的一致性
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        // 4. 设置 checkpoint 的存储目录
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop162:8020/gmall/" + ckAndGroupIdAndJobName);
        // 5. 设置 checkpoint 的超时时间. 超过这个时间, 放弃这次 checkpoint
        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000);
        // 6. checkpoint 的并发数: 同时最多执行一个 checkpoint
        // env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);  // 可以省略
        // 7. 两个 checkpoint 的之间最小时间间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
        // 8. 设置 checkpoint 的保存策略
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(RETAIN_ON_CANCELLATION);
        
        
        KafkaSource<String> source = FlinkSourceUtil.getKafkaSource(topic, ckAndGroupIdAndJobName);
        DataStreamSource<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");
        
        // 进行业务处理.调用子类的一个方法
        handle(env, stream);
        
        try {
            // 给 job 设置 name
            //            env.execute(ckAndGroupIdAndJobName);
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        
    }
}
